package com.lmq
import com.lmq.utils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{DoubleType, IntegerType, LongType, StringType, StructField, StructType}
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * TODO:
 *      第一步预处理:
 *      添加一列 从日期字符串转为Double的时间戳Time
 *      最后添加一列TimeRank，即在会话内按照时间戳Time升序后的排名。
 */
object preprocess {

  Logger.getLogger("org.apache.spark")
    .setLevel(Level.WARN)

  val spark = SparkSession.builder()
    .master("local[*]")
    .appName("Test")
    .getOrCreate()


  def main(args: Array[String]): Unit = {
    val utils = new utils()

    val schema = StructType(
      Array(
        StructField("SessionId", IntegerType, nullable = true),
        StructField("TimeStr", StringType, nullable = true),
        StructField("ItemId", IntegerType, nullable = true),
        StructField("Context", StringType, nullable = true)
      )
    )
    val df: DataFrame = spark.read
      .schema(schema)
      .option("header",value = true)
      .csv("file:///home/iptv/yoochoose/yoochoose-clicks.dat")
//      .csv("D:\\pythonProject\\pythonProject\\ComparisonWithYoochoose\\src\\yoochoose-clicks.csv")
    df.show(false)

    val timeTransfer = udf((x:String)=>  utils.timeStr2Tsp(x) )
    df.select(
      col("SessionId"),
      col("TimeStr"),
      col("ItemId"),
      col("Context"),
      timeTransfer( col("TimeStr") ).alias("Time")
    )
    .createOrReplaceTempView("allData")

    val v: DataFrame = spark.sql(""" select * ,rank() over (partition by SessionId order by Time) as
    `TimeRank` from allData""")
    v.show(false)
//    v.createTempView("V")
//    // remove sessions which the length is less than 1 and apearance is less than 5.
//    spark.sql(
//      """select v.* from
//        |(select ItemId from
//        |( select ItemId, count(1) as cnt
//        |from v) A where cnt>=5) B
//        |left join v
//        |on B.ItemId=v.ItemId
//        |""".stripMargin)



    v.coalesce(1).write
      .option("header","false")
      .csv("file:///home/iptv/yoochoose/Reslt.csv")

//      .toPandas()\
//    .to_csv("helloResult.csv",index=None)







  }

}
